package com.sina.rd.dsp.server;

import com.netflix.curator.framework.api.CuratorWatcher;
import com.sina.rd.dsp.nedis.NedisService;
import com.sina.rd.dsp.util.Tools;
import com.sina.rd.dsp.zookeeper.ZKProxy;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hbase.util.Sleeper;
import org.apache.hadoop.net.DNS;
import org.apache.zookeeper.WatchedEvent;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.JedisPool;

import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

/**
 * Created with IntelliJ IDEA.
 * User: xinrong3
 * Date: 14-4-2
 * Time: 下午8:01
 *
 * NedisService
 */
public class SimpleServer implements SCServer{
    public static final Log LOG = LogFactory.getLog(SimpleServer.class);
    /*
* Strings to be used in forming the exception message for
* RegionsAlreadyInTransitionException.
*/
    protected static final String OPEN = "OPEN";
    protected static final String CLOSE = "CLOSE";
    protected static final String CLUSTERNAME = "cluster.name";
    protected static final String ZK_SEPERATE="/";
    public static final String ZOOKEEPER_QUORUM = "hbase.zookeeper.quorum";
    public static final String ROUTE_MAP_KEY="route.map.key";

    public  final String CluserName;

    public final String ServerPathInZK;

    public final String RoutePahtInZK;

    private final int msgInterval;
    private final Sleeper sleeper;

    private RouteMap rMap=new RouteMap();

    protected volatile boolean stopped = false;
    protected volatile boolean abortRequested;

    private ZKProxy zkProxy;
    private Properties prop;
    private String localServerPath;

    private RouteWatcher watcher=new RouteWatcher();

    public SimpleServer(Properties properties) {
        //configure
        prop=properties;
        //proxy
        zkProxy=new ZKProxy(properties.getProperty(ZOOKEEPER_QUORUM));
        //path in zk
        CluserName=prop.getProperty(CLUSTERNAME,"scserver-cluster");
        ServerPathInZK=ZK_SEPERATE+CluserName+ZK_SEPERATE+"server-pool";
        RoutePahtInZK=ZK_SEPERATE+CluserName+ZK_SEPERATE+"route";

        //interrupt configure
        this.msgInterval = Integer.parseInt(prop.getProperty("hbase.regionserver.msginterval", "3000"));
        this.sleeper = new Sleeper(this.msgInterval, this);
    }

    /**
     * 从固定地址拉取route表
     * @param routeServer master
     */
    private boolean fetchRoute(String routeServer) {
        if(Tools.notNullOrEmpty(routeServer)){
            String part[]=routeServer.split(":");
            try {
                int jedisport=Integer.parseInt(part[1]);
                Jedis jedis=new Jedis(part[0],jedisport);
                Map<String,String> map=jedis.hgetAll(ROUTE_MAP_KEY);
                for(Map.Entry<String,String> e:map.entrySet()){
                    rMap.resetBidRoute(e.getKey(),jedis.hgetAll(e.getKey()));
                }
                return true;
            }catch (Exception e){
                LOG.error("route error,wrong cached key");
            }
        }else {
            LOG.info("master route is not ready");
        }
        return false;
    }

    private void startRPCHandlerThreads() {
        NedisService service=new NedisService(prop,rMap);
        service.start();
    }

    /**
     * 启动时拉取route信息，监视 zk中route的变动，若有变动则重新获取rout
     */
    private void startTraceRouteChangement() {
        String routeServer=zkProxy.geDatatAndWatch(this.RoutePahtInZK,watcher);
        while (!fetchRoute(routeServer)){
            LOG.info("fetch route failed,try later");
            try {
                Thread.sleep(300l);
            } catch (InterruptedException e) {
                LOG.debug("fetch rout is interrupted");
            }
        }
    }


    private void createMyEphemeralNode() throws UnknownHostException {
        zkProxy.create(getLocalServerPath());
    }

    @Override
    public void run() {
        prepareServiceThread();

        // We registered with the Master.  Go into run mode.
        long lastMsg = 0;
        long oldRequestCount = -1;

        while (!isStopped()){
            long now = System.currentTimeMillis();
            if ((now - lastMsg) >= 10*1000) {
//                tryServerReport(lastMsg, now);
                lastMsg = System.currentTimeMillis();
            }
            if (!this.stopped) this.sleeper.sleep();
        }
    }

    private boolean prepareServiceThread() {
        int tryTime=0;
        while(tryTime++<10){
            try {
                createMyEphemeralNode();
                break;
            } catch (Exception e) {
                LOG.error(e);
                tryTime++;
            }
        }
        //启动追踪route表的线程
        startTraceRouteChangement();


        //启动Netty线程
        startRPCHandlerThreads();

        return true;
    }


    @Override
    public void stop(String why) {
        this.stopped = true;
        LOG.info("STOPPED: " + why);
        // Wakes run() if it is sleeping
        sleeper.skipSleepCycle();
    }

    @Override
    public boolean isStopped() {
        return  stopped;
    }

    public  String getLocalServerPath() throws UnknownHostException {
        String hostName=ServerPathInZK;
        hostName+=ZK_SEPERATE+DNS.getDefaultHost("default","default");
        return  hostName;
    }


    private class RouteWatcher implements CuratorWatcher{
        @Override
        public void process(WatchedEvent event) throws Exception {
            String routeServer=zkProxy.geDatatAndWatch(RoutePahtInZK,this);
            fetchRoute(routeServer);
        }
    }


    public static void main(String ...args){
        Properties p=new Properties();
        p.setProperty(ZOOKEEPER_QUORUM,"h011171.hebe.grid.sina.com.cn:2181,h011200.hebe.grid.sina.com.cn:2181,h011196.hebe.grid.sina.com.cn:2181");
        SimpleServer server=new SimpleServer(p);
        server.run();
    }
}
